package com.seaboxdata.interfaceJob.queuing;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.seaboxdata.entity.queuing.TrafficGetLineBrandInformation;
import com.seaboxdata.kafka.BaseStringKafkaProducer;
import com.seaboxdata.service.queuing.TrafficGetLineBrandInformationService;
import java.io.IOException;
import java.math.BigInteger;
import java.net.URI;
import java.net.URISyntaxException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
@Configuration
@Slf4j
public class TrafficGetLineBrandInformationJob {

    @Value("${Traffic.TrafficGetLineBrandInformation.url}")
    private String url;

    @Value("${Traffic.TrafficGetLineBrandInformation.token}")
    private String token;

    @Value("${Traffic.TrafficGetLineBrandInformation.Topic}")
    private String topic;
    @Autowired
    private BaseStringKafkaProducer baseStringKafkaProducer;

    @Value("${interface.isTest}")
    private String isTest;

    @Autowired
    private TrafficGetLineBrandInformationService trafficGetLineBrandInformationService;

    @Scheduled(cron = "${Traffic.TrafficGetLineBrandInformation.Scheduled}")//每天执行一次
    public void writeKafka() {

        //如果是测试,直接退出
        if ("true".equals(isTest)) {
            return;
        }
        String dateStr = Long.toString(System.currentTimeMillis() / 1000L);
        String stringToMD5 = stringToMD5(token + dateStr);
        try {
            List<TrafficGetLineBrandInformation> trafficGetLineBrandInformations = send(url, stringToMD5, dateStr,
                    TrafficGetLineBrandInformation.class);
            if (CollectionUtils.isNotEmpty(trafficGetLineBrandInformations)) {
                trafficGetLineBrandInformationService.saveOrUpdateBatch(trafficGetLineBrandInformations);
            }
//            if (!CollectionUtils.isEmpty(trafficGetLineBrandInformations)) {
//                //写入数据到kafka,入库在本地消费
//                Date date = new Date();
//                for (TrafficGetLineBrandInformation trafficGetLineBrandInformation : trafficGetLineBrandInformations) {
//                    trafficGetLineBrandInformation.setCjsj(date);
//                    System.out.println(trafficGetLineBrandInformation);
//                    baseStringKafkaProducer.sendMessageToKafka(JSONObject.toJSONString(trafficGetLineBrandInformation), topic);
//                }
//            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public String stringToMD5(String plainText) {
        byte[] secretBytes = null;
        try {
            secretBytes = MessageDigest.getInstance("md5").digest(
                    plainText.getBytes());
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException("没有这个md5算法！");
        }
        String md5code = new BigInteger(1, secretBytes).toString(16);
        for (int i = 0; i < 32 - md5code.length(); i++) {
            md5code = "0" + md5code;
        }
        return md5code;
    }

    public List send(String url, String token, String time, Class<?> cls) throws IOException, URISyntaxException {
        // 创建Httpclient对象
        CloseableHttpClient httpclient = HttpClients.createDefault();
        // 定义请求的参数
        URI uri = new URIBuilder(url).setParameter("s_key", "SZDP").setParameter("s_sign", token)
                .setParameter("s_time", time).build();
        List<Object> results = new ArrayList<>();
        // 创建http GET请求
        HttpGet httpGet = new HttpGet(uri);
        //response 对象
        CloseableHttpResponse response = null;
        try {
            Date timelocal = new Date();
            // 执行http get请求
            response = httpclient.execute(httpGet);
            // 判断返回状态是否为200
            if (response.getStatusLine().getStatusCode() == 200) {
                String result = EntityUtils.toString(response.getEntity(), "UTF-8");
                JSONArray speeds = null;
                JSONObject next = (JSONObject) JSONObject.parse(result);
                Object t = next.get("data");
                speeds = (JSONArray) t;
                Iterator<Object> iterator1 = speeds.iterator();
                while (iterator1.hasNext()) {
                    JSONObject next1 = (JSONObject) iterator1.next();
                    TrafficGetLineBrandInformation trafficGetLineBrandInformation = (TrafficGetLineBrandInformation) next1
                            .toJavaObject(cls);
                    trafficGetLineBrandInformation.setCjsj(timelocal);
                    results.add(trafficGetLineBrandInformation);
                }
            }
        } catch (Exception e) {

            log.info("");
        } finally {
            if (response != null) {
                response.close();
            }
            httpclient.close();
        }
        return results;
    }


}
